package action

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object E1_Action_topN {

  def main(args: Array[String]): Unit = {
    topN()
  }

  """
    |（1）品类是指产品的分类，大型电商网站品类分多级，咱们的项目中品类只有一级，不同的公司可能对热门的定义不一样。我们按照每个品类的点击、下单、支付的量来统计热门品类。
    |鞋 点击数 下单数 支付数
    |衣服 点击数 下单数 支付数
    |电脑 点击数 下单数 支付数
    |
    |""".stripMargin

  def topN() = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("action")
    val sc = new SparkContext(sparkConf)
    val textRdd = sc.textFile("file:///D:\\workspace\\lab\\learnbigdata\\learnspark\\doc\\spark-sql\\data\\user_visit_action.txt")
    // 将数据转换结构
    //   点击的场合 : ( 品类ID，( 1, 0, 0 ) )
    //   下单的场合 : ( 品类ID，( 0, 1, 0 ) )
    //   支付的场合 : ( 品类ID，( 0, 0, 1 ) )
    val flatRDD: RDD[(String, (Int, Int, Int))] = textRdd.flatMap(action => {
      val datas = action.split(",")
      if (datas(6) != "-1") {
        // 点击场合
        List((datas(6), (1, 0, 0)))
      } else if (datas(8) != "null") {
        // 下单的场合
        val ids = datas(8).split(",")
        ids.map(id => (id, (0, 1, 0)))
      } else if (datas(10) != "null") {
        // 支付的场合
        val ids = datas(10).split(",")
        ids.map(id => (id, (0, 0, 1)))
      } else {
        Nil
      }
    })
    // 将相同的品类ID的数据进行分组聚合
    //  (品类ID，(点击数量, 下单数量, 支付数量))
    val analysisRDD: RDD[(String, (Int, Int, Int))] = flatRDD.reduceByKey((t1, t2) => {
      (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
    })
    // 按照点击数、下单数、支付数
    val sortedCategories = analysisRDD.sortBy(_._2, false)
    //    val sortedCategories = analysisRDD.sortBy(_._2._1, false)
    //    val sortedCategories = analysisRDD.sortBy(_._2._2, false)
    //    val sortedCategories = analysisRDD.sortBy(_._2._3, false)

    // 按照综合排名进行排序
    val sortedCategoriesV2 = analysisRDD.sortBy(categoryCount => {
      val (clickCount, orderCount, payCount) = categoryCount._2
      clickCount * 0.2 + orderCount * 0.3 + payCount * 0.5
    }, ascending = false)

    var top10Categories = sortedCategories.take(10)
    top10Categories.foreach {
      case (category, (clickcount, ordercount, paycount)) =>
        println(s"品类: $category, 点击数: $clickcount, 下单数: $ordercount, 支付数: $paycount")
    }
    sc.stop
  }
}
